package com.wdl.networkflow.topN

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor, MapState, MapStateDescriptor}
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

import java.sql.Timestamp
import java.util
import java.util.Map
import scala.collection.mutable.ListBuffer

class TopNHotPage(n: Int) extends KeyedProcessFunction[Long, PageViewCount, String] {

  lazy val pageCountMapState: MapState[String, Long] = getRuntimeContext.getMapState(
    new MapStateDescriptor[String, Long]("page-count-map", classOf[String], classOf[Long]))

  override def processElement(value: PageViewCount,
                              ctx: KeyedProcessFunction[Long, PageViewCount, String]#Context,
                              out: Collector[String]): Unit = {
    pageCountMapState.put(value.url ,value.count)

    ctx.timerService().registerEventTimeTimer(value.windowEnd + 1)

    /** 注册定时器，用于定时清空 pageCountListState */
    ctx.timerService().registerEventTimeTimer(value.windowEnd + (60 * 1000L))
  }

  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[Long, PageViewCount, String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    val allPageCountList: ListBuffer[(String, Long)] = ListBuffer()

    val iter: util.Iterator[Map.Entry[String, Long]] = pageCountMapState.entries().iterator()

    while (iter.hasNext){
      val key: String = iter.next().getKey
      val value: Long = iter.next().getValue
      allPageCountList += ((key, value))
    }

    if (timestamp == ctx.getCurrentKey + (60 * 1000L)){
      pageCountMapState.clear()
      return
    }

    val pageViewCounts: ListBuffer[(String, Long)] = allPageCountList.sortWith(_._2 > _._2).take(n)

    val result: StringBuilder = new StringBuilder
    result.append("时间：").append( new Timestamp(timestamp - 1) ).append("\n")
    // 遍历sorted列表，输出TopN信息
    for( i <- pageViewCounts.indices ){
      // 获取当前商品的count信息
      val currentItemCount = pageViewCounts(i)
      result.append("Top").append(i + 1).append(":")
        .append(" 页面url=").append(currentItemCount._1)
        .append(" 访问量=").append(currentItemCount._2)
        .append("\n")
    }
    result.append("==============================\n\n")

    out.collect(result.toString())

  }
}
